#ifndef _PARALLELSOCKETACCEPTOR_HPP_
#define _PARALLELSOCKETACCEPTOR_HPP_

#include "jhdefs.hpp"

NET_NAMESPACE_BEGIN

template <class ServiceHandler, class SR>
class TParallelSocketAcceptor
{
public:
	typedef TParallelSocketReactor<SR> ParallelReactor;

	explicit TParallelSocketAcceptor(CServerSocket& socket,
		unsigned threads = CEnvironment::processorCount()):
	m_socket(socket),
		m_pReactor(0),
		m_threads(threads),
		m_next(0)
		/// Creates a ParallelSocketAcceptor using the given ServerSocket, 
		/// sets number of threads and populates the reactors vector.
	{
		init();
	}

	TParallelSocketAcceptor(CServerSocket& socket,
		CSocketReactor& reactor,
		unsigned threads = CEnvironment::processorCount()):
	m_socket(socket),
		m_pReactor(&reactor),
		m_threads(threads),
		m_next(0)
	{
		init();
		m_pReactor->addEventHandler(m_socket,
			TObserver<ParallelSocketAcceptor,
			CReadableNotification>(*this, &TParallelSocketAcceptor::onAccept));
	}

	virtual ~TParallelSocketAcceptor()
	{
		try
		{
			if (m_pReactor)
			{
				m_pReactor->removeEventHandler(m_socket,
					TObserver<ParallelSocketAcceptor,
					CReadableNotification>(*this, &TParallelSocketAcceptor::onAccept));
			}
		}
		catch (...)
		{
			jh_unexpected();
		}
	}

	void setReactor(CSocketReactor& reactor)
	{
		m_pReactor = &reactor;
		if (!m_pReactor->hasEventHandler(m_socket, 
			TObserver<ParallelSocketAcceptor,
			CReadableNotification>(*this, &TParallelSocketAcceptor::onAccept)))
		{
			registerAcceptor(reactor);
		}
	}

	virtual void registerAcceptor(CSocketReactor& reactor)
	{
		if (m_pReactor)
			throw InvalidAccessException("Acceptor already registered.");

		m_pReactor = &reactor;
		m_pReactor->addEventHandler(m_socket,
			TObserver<ParallelSocketAcceptor,
			CReadableNotification>(*this, &TParallelSocketAcceptor::onAccept));
	}

	virtual void unregisterAcceptor()
	{
		if (m_pReactor)
		{
			m_pReactor->removeEventHandler(m_socket,
				TObserver<ParallelSocketAcceptor,
				CReadableNotification>(*this, &TParallelSocketAcceptor::onAccept));
		}
	}

	void onAccept(CReadableNotification* pNotification)
	{
		pNotification->release();
		CStreamSocket sock = m_socket.acceptConnection();
		m_pReactor->wakeUp();
		createServiceHandler(sock);
	}

protected:
	virtual ServiceHandler* createServiceHandler(CStreamSocket& socket)
		/// Create and initialize a new ServiceHandler instance.
		///
		/// Subclasses can override this method.
	{
		std::size_t next = m_next++;
		if (m_next == _reactors.size()) m_next = 0;
		m_reactors[next]->wakeUp();
		return new ServiceHandler(socket, *m_reactors[next]);
	}

	CSocketReactor* reactor()
	{
		return m_pReactor;
	}

	CSocket& socket()
		/// Returns a reference to the SocketAcceptor's socket.
	{
		return m_socket;
	}

	void init()
		/// Populates the reactors vector.
	{
		jh_assert (m_threads > 0);

		for (unsigned i = 0; i < m_threads; ++i)
			m_reactors.push_back(new ParallelReactor);
	}

	typedef std::vector<typename ParallelReactor::Ptr> ReactorVec;

	ReactorVec& reactors()
	{
		return m_reactors;
	}

	CSocketReactor* reactor(std::size_t idx)
	{
		return m_reactors.at(idx).get();
	}

	std::size_t& next()
	{
		return m_next;
	}

private:
	TParallelSocketAcceptor();
	TParallelSocketAcceptor(const TParallelSocketAcceptor&);
	TParallelSocketAcceptor& operator = (const TParallelSocketAcceptor&);

	CServerSocket   m_socket;
	CSocketReactor* m_pReactor;
	unsigned       m_threads;
	ReactorVec     m_reactors;
	std::size_t    m_next;
};

NET_NAMESPACE_END

#endif
